/*
 * Copyright (c) Facebook, Inc. and its affiliates.
 *
 * Licensed under the Apache License Version 2.0 with LLVM Exceptions
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *   https://llvm.org/LICENSE.txt
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#pragma once

#include <unifex/bind_back.hpp>
#include <unifex/just_done.hpp>
#include <unifex/manual_lifetime.hpp>
#include <unifex/receiver_concepts.hpp>
#include <unifex/scope_guard.hpp>
#include <unifex/sender_concepts.hpp>
#include <unifex/stream_concepts.hpp>

#include <algorithm>
#include <optional>
#include <type_traits>

#include <unifex/detail/prologue.hpp>

namespace unifex {
namespace _single {
template <typename Sender, typename Receiver>
struct _next_op {
  struct type;
};
template <typename Sender, typename Receiver>
using next_operation =
    typename _next_op<Sender, remove_cvref_t<Receiver>>::type;

template <typename Sender, typename Receiver>
struct _next_op<Sender, Receiver>::type {
  union {
    Receiver receiver_;
    manual_lifetime<connect_result_t<Sender, Receiver>> innerOp_;
  };
  bool done_;

  template(typename Receiver2)  //
      (requires constructible_from<
          Receiver,
          Receiver2>)  //
      explicit type(Receiver2&& receiver)
    : receiver_((Receiver2 &&) receiver)
    , done_(true) {}

  explicit type(Sender&& sender, Receiver&& receiver) : done_(false) {
    unifex::activate_union_member_with(innerOp_, [&] {
      return unifex::connect(
          static_cast<Sender&&>(sender), (Receiver &&) receiver);
    });
  }

  ~type() {
    if (done_) {
      receiver_.~Receiver();
    } else {
      unifex::deactivate_union_member(innerOp_);
    }
  }

  void start() noexcept {
    if (done_) {
      unifex::set_done(std::move(receiver_));
    } else {
      unifex::start(innerOp_.get());
    }
  }
};

template <typename Sender>
struct _stream {
  struct type;
};
template <typename Sender>
using stream = typename _stream<remove_cvref_t<Sender>>::type;

template <typename Sender>
struct _stream<Sender>::type {
  std::optional<Sender> sender_;

  struct next_sender {
    std::optional<Sender> sender_;

    template <
        template <typename...>
        class Variant,
        template <typename...>
        class Tuple>
    using value_types = sender_value_types_t<Sender, Variant, Tuple>;

    template <template <typename...> class Variant>
    using error_types = sender_error_types_t<Sender, Variant>;

    static constexpr bool sends_done = true;

    static constexpr blocking_kind blocking =
        std::min(sender_traits<Sender>::blocking(), blocking_kind::maybe());

    static constexpr bool is_always_scheduler_affine =
        sender_traits<Sender>::is_always_scheduler_affine;

    friend constexpr blocking_kind
    tag_invoke(tag_t<unifex::blocking>, const next_sender& s) noexcept {
      if (s.sender_) {
        return unifex::blocking(*s.sender_);
      } else {
        return blocking_kind::always_inline;
      }
    }

    template <typename Receiver>
    auto connect(Receiver&& receiver) {
      if (sender_) {
        return next_operation<Sender, Receiver>{
            *std::move(sender_), (Receiver &&) receiver};
      } else {
        return next_operation<Sender, Receiver>{(Receiver &&) receiver};
      }
    }
  };

  friend next_sender tag_invoke(tag_t<next>, type& s) {
    scope_guard g{[&]() noexcept {
      s.sender_.reset();
    }};
    return next_sender{std::move(s.sender_)};
  }

  friend auto tag_invoke(tag_t<cleanup>, type&) noexcept { return just_done(); }

  template <typename Sender2>
  explicit type(Sender2&& sender)
    : sender_(std::in_place, (Sender2 &&) sender) {}
};
}  // namespace _single

namespace _single_cpo {
inline const struct _fn {
  template <typename Sender>
  auto operator()(Sender&& sender) const {
    return _single::stream<Sender>{(Sender &&) sender};
  }
  constexpr auto operator()() const
      noexcept(std::is_nothrow_invocable_v<tag_t<bind_back>, _fn>)
          -> bind_back_result_t<_fn> {
    return bind_back(*this);
  }
} single{};
}  // namespace _single_cpo

using _single_cpo::single;

}  // namespace unifex

#include <unifex/detail/epilogue.hpp>
